import json
import sys

from PyQt5.QtCore import QThread, pyqtSignal
from PyQt5.QtGui import QTextCursor
from kafka import KafkaConsumer, KafkaProducer

from PyQt5 import QtWidgets


class ProducerGui(QtWidgets.QWidget):
    """KAFKA消费者"""

    def __init__(self):
        super().__init__()

        self.setWindowTitle("KAFKA生产者模拟工具")

        self.host = QtWidgets.QLineEdit()
        self.host.setPlaceholderText("KAFKA服务器地址")

        self.port = QtWidgets.QLineEdit()
        self.port.setPlaceholderText("端口")

        self.topic = QtWidgets.QLineEdit()
        self.topic.setPlaceholderText("TOPIC")

        self.connect_button = QtWidgets.QPushButton("发送", )
        self.close_button = QtWidgets.QPushButton("清空消息")

        self.receive_msg_edit = QtWidgets.QTextEdit()
        self.receive_msg_edit.setReadOnly(True)

        self.timed_transmission = QtWidgets.QCheckBox()
        self.sending_frequency = QtWidgets.QLineEdit()
        self.sending_frequency.setPlaceholderText("1")

        layout1 = QtWidgets.QHBoxLayout()
        layout1.addWidget(QtWidgets.QLabel("地址："))
        layout1.addWidget(self.host)
        layout1.addWidget(QtWidgets.QLabel("端口："))
        layout1.addWidget(self.port)
        layout1.addWidget(QtWidgets.QLabel("TOPIC："))
        layout1.addWidget(self.topic)
        layout1.addWidget(self.connect_button)
        layout1.addWidget(self.close_button)

        layout2 = QtWidgets.QVBoxLayout()
        layout2.addWidget(QtWidgets.QLabel("接收到的消息："))
        layout2.addWidget(self.receive_msg_edit)

        layout3 = QtWidgets.QVBoxLayout()
        layout3.addLayout(layout1)
        layout3.addLayout(layout2)
        self.setLayout(layout3)


class KafkaProducerThread(QThread):
    message_received = pyqtSignal(str)

    def __init__(self, kafka_url, topic):
        super().__init__()

        self.topic = topic
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_url,
            value_serializer=lambda m: json.dumps(m).encode('ascii'), )  # json 格式化发送的内容

    def run(self):
        print(self.producer)


class KProducer:
    # 生产者类
    def __init__(self, bootstrap_servers, topic):
        """
        kafka 生产者
        :param bootstrap_servers: 地址
        :param topic:  topic
        """
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda m: json.dumps(m).encode('ascii'), )  # json 格式化发送的内容
        self.topic = topic

    def sync_producer(self, data_li: list):
        """
        同步发送 数据
        :param data_li:  发送数据
        :return:
        """
        for data in data_li:
            future = self.producer.send(self.topic, data)
            record_metadata = future.get(timeout=10)  # 同步确认消费
            partition = record_metadata.partition  # 数据所在的分区
            offset = record_metadata.offset  # 数据所在分区的位置
            print('save success, partition: {}, offset: {}'.format(partition, offset))

    def async_producer(self, data_li: list):
        """
        异步发送数据
        :param data_li:发送数据
        :return:
        """
        for data in data_li:
            self.producer.send(self.topic, data)
        self.producer.flush()  # 批量提交

    def async_producer_callback(self, data_li: list):
        """
        异步发送数据 + 发送状态处理
        :param data_li:发送数据
        :return:
        """
        for data in data_li:
            self.producer.send(self.topic, data).add_callback(self.send_success).add_errback(self.send_error)
        self.producer.flush()  # 批量提交

    def send_success(self, *args, **kwargs):
        """异步发送成功回调函数"""
        print('save success')
        return

    def send_error(self, *args, **kwargs):
        """异步发送错误回调函数"""
        print('save error')
        return

    def close_producer(self):
        try:
            self.producer.close()
        except:
            pass


if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    window = ProducerGui()
    window.show()
    sys.exit(app.exec_())
